package com.yzh.luckydraw.application.mq.consumer;

import cn.hutool.core.lang.Assert;
import com.alibaba.fastjson.JSON;
import com.yzh.luckydraw.application.mq.producer.KafkaProducer;
import com.yzh.luckydraw.common.Constants;
import com.yzh.luckydraw.domain.activity.model.vo.InvoiceVO;
import com.yzh.luckydraw.domain.award.model.req.GoodsReq;
import com.yzh.luckydraw.domain.award.model.res.DistributionRes;
import com.yzh.luckydraw.domain.award.service.factory.DistributionGoodsFactory;
import com.yzh.luckydraw.domain.award.service.goods.IDistributionGoods;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Optional;

/**
 * 中奖发货监听消息
 */
@Component
public class InvoiceListener {
    private Logger logger = LoggerFactory.getLogger(InvoiceListener.class);
    @Resource
    private DistributionGoodsFactory distributionGoodsFactory;

    @KafkaListener(topics = KafkaProducer.TOPIC_INVOICE, groupId = "luckydraw")
    public void onMessage(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        Optional<?> value = Optional.ofNullable(record.value());
        //1.消息是否存在
        if (!value.isPresent()) {
            return;
        }
        //2.处理MQ消息
        try {
            //1.转化对象
            InvoiceVO invoiceVO = JSON.parseObject((String) value.get(), InvoiceVO.class);

            //2.获取发送奖品工厂，执行发奖
            IDistributionGoods distributionGoodsService = distributionGoodsFactory.getDistributionGoodsService(invoiceVO.getAwardType());
            DistributionRes distributionRes = distributionGoodsService.doDistribution(new GoodsReq(invoiceVO.getuId(), invoiceVO.getOrderId(), invoiceVO.getAwardId(), invoiceVO.getAwardName(), invoiceVO.getAwardContent()));
            
            Assert.isTrue(Constants.AwardState.SUCCESS.getCode().equals(distributionRes.getCode()), distributionRes.getInfo());
            
            //3.打印日志
            logger.info("消费MQ消息成功，topic：{} uId：{} 发奖结果：{}", topic, invoiceVO.getuId(), JSON.toJSONString(distributionRes));
            
            //4.消息消费完成
            ack.acknowledge();
        }catch (Exception e) {
            //发奖环节失败，消息重试。所有环节，发货、更新库，都要保证幂等性
            logger.error("消费MQ消息，失败 topic：{} msg：{}", topic, value.get());
            throw e;
        }
    }
}
